package com.cskaoyan.mall.promo.mq.producer;

import com.alibaba.fastjson.JSON;
import com.cskaoyan.mall.order.dto.CreateSeckillOrderRequest;
import com.cskaoyan.mall.promo.cache.CacheManager;
import com.cskaoyan.mall.promo.dal.persistence.PromoItemMapper;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.Lock;

@Component
public class PromoOrderProducer {

    TransactionMQProducer producer;
    @Autowired
    PromoItemMapper promoItemMapper;

    @Autowired
    CacheManager cacheManager;

    @Autowired
    RedissonClient redissonClient;

    @PostConstruct
    public void init() {
        // 创建producer对象
        producer = new TransactionMQProducer("promo_order_producer44");

        // 设置nameserver地址
        producer.setNamesrvAddr("127.0.0.1:9876");

        // 设置事务监听器
        producer.setTransactionListener(new TransactionListener() {

            /*
                    实现秒杀服务的本地事务
             */
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                Map<String, Long> paramMap = (Map<String, Long>) arg;

                Long productId = paramMap.get("productId");
                Long psId = paramMap.get("psId");
                // 加锁
                Integer effetiveRow = 0;
//                 无法解决超卖问题
//                synchronized (this) {
//                    effetiveRow = promoItemMapper.decreaseStock(productId, psId);
//                }

                // 释放锁

                // 使用基于Redis实现的分布式锁

                // 1. 获取这把锁
                String lockKey = "promo_decrease_lock: " + psId + ":" + productId;
                RLock lockObj = redissonClient.getLock(lockKey);
                // 2. 加锁
                lockObj.lock();
                try {
                    effetiveRow = promoItemMapper.decreaseStock(productId, psId);
                } finally {
                    // 3. 释放锁
                    lockObj.unlock();
                }

                String localTransactionKey = "promo_order_local_transaction: " + msg.getTransactionId();
                if (effetiveRow < 1) {
                   // 商品售罄, 秒杀失败
                    cacheManager.setCache(localTransactionKey, "ROLLBACK_MESSAGE", 1);
                    String stockNotEnoughKey = "promo_stock_not_enough: "+psId+":"+productId;
                    cacheManager.setCache(stockNotEnoughKey,"none",1);
                  return  LocalTransactionState.ROLLBACK_MESSAGE;
                }

                //  本地事务执行成功
                cacheManager.setCache(localTransactionKey, "COMMIT_MESSAGE", 1);
                return LocalTransactionState.COMMIT_MESSAGE;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                String localTransactionKey = "promo_order_local_transaction: " + msg.getTransactionId();

                String transactionResult = cacheManager.checkCache(localTransactionKey);
                if (transactionResult == null || transactionResult.isEmpty()) {
                    return LocalTransactionState.UNKNOW;
                }

                if ("ROLLBACK_MESSAGE".equals(transactionResult)) {
                    // 本地事务执行失败
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }

                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });

        // 启动producer
        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }

    }





    public boolean sendPromoOrderMessage(CreateSeckillOrderRequest request, Long productId, Long psId) {

        String parmJson = JSON.toJSONString(request);

        HashMap<String, Long> transactionParam = new HashMap<>();
        transactionParam.put("productId", productId);
        transactionParam.put("psId", psId);

        // 构造待发送的消息
        Message message = new Message();
        message.setTopic("promo_order");
        message.setBody(parmJson.getBytes(Charset.forName("utf-8")));

        //transactionSendResult.getLocalTransactionState() 获取本地事务的执行结果
        TransactionSendResult transactionSendResult = null;
        try {
           transactionSendResult
                   // sendMessageInTransaction 方法的第二个参数，
                   // 传递给executeLocalTransaction，即给本地事务执行所使用的参数
                    = producer.sendMessageInTransaction(message, transactionParam);
        } catch (MQClientException e) {
            e.printStackTrace();
        }

        if (transactionSendResult != null && LocalTransactionState.COMMIT_MESSAGE.equals(transactionSendResult.getLocalTransactionState())) {
            //  事务消息发送成功，本地事务执行成功 ——> 分布式事务执行成功
            return true;
        }

        // 1. 发送事务消息失败返回false
        // 2. 本地事务执行失败
        return  false;
    }
}
